package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.OrderDetail;
import com.atguigu.gmall.realtime.bean.OrderInfo;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.utils.DateUtils;
import com.atguigu.gmall.realtime.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 订单和订单明细关联代码实现
 * <p>
 * -需要启动的进程<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp zk、kafka、maxwell、hdfs、hbase、redis、BaseDBApp、OrderWideApp<br>
 * -执行流程<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > 运行模拟生成业务数据的jar<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > 会向业务数据库MySQL中插入生成业务数据<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > MySQL会将变化的数据放到Binlog中<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > Maxwell从 binlog 中获取数据，将数据封装为 json 字符串发送到 kafka 的 ods 主题  ods_base_db_m<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > BaseDBApp 从 ods_base_db_m 主题中读取数据，进行分流<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp &事实----写回到kafka的dwd主题<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp &维度----保存到phoenix的维度表中<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > OrderWideApp 从 dwd 主题中获取订单和订单明细数据<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > 使用 intervalJoin 对订单和订单明细进行双流 join<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp > 将用户维度关联到订单宽表上<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp 基本的维度关联<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp 优化1：旁路缓存的优化<br>
 * &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp&nbsp 优化2：异步IO<br>
 *
 * @author lvbingbing
 * @date 2022-03-27 20:06
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        // 1. 基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度设置
        env.setParallelism(4);
        // 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 开启外部化检查点，作业取消时保留检查点
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 6000L));
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck/gmall"));
        // 2.6 指定操作 hdfs 用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3. 从 kafka 读取数据，做结构转换，并指定时间戳和 watermark
        String groupId = "order_wide_app_group";
        SingleOutputStreamOperator<OrderInfo> orderInfoDs = env.addSource(KafkaUtils.getKafkaSource("dwd_order_info", groupId))
                // 结构转换
                .map(new MapOrderInfo())
                // 指定时间戳和 watermark
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new OrderInfoTimestampAssigner()));
        SingleOutputStreamOperator<OrderDetail> orderDetailDs = env.addSource(KafkaUtils.getKafkaSource("dwd_order_detail", groupId))
                // 结构转换
                .map(new MapOrderDetailInfo())
                // 指定时间戳和 watermark
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new OrderDetailInfoTimestampAssigner()));
        // 4. 双流join，使用intervalJoin
        SingleOutputStreamOperator<OrderWide> orderWideDs = orderInfoDs.keyBy(OrderInfo::getId)
                .intervalJoin(orderDetailDs.keyBy(OrderDetail::getOrder_id))
                // 设置正负5秒，防止在业务系统中主表与从表保存的时间差
                .between(Time.seconds(-5), Time.seconds(5))
                // 处理连接后的数据
                .process(new OrderInfoProcessJoinFunction());
        // 5. 异步关联维度信息
        SingleOutputStreamOperator<OrderWide> orderWideWithDimInfo = asyncJoinDimInfo(orderWideDs);
        // 6、将订单宽表数据写回到 kafka 的 dwm 主题
        orderWideWithDimInfo.map(JSON::toJSONString)
                .addSink(KafkaUtils.getKafkaSinkAtLeastOnce("dwm_order_wide"));
        // 7、触发程序执行
        env.execute();
    }

    /**
     * 异步关联维度信息
     *
     * @param orderWideDs 待进行维度关联的订单信息
     * @return 维度关联后的订单宽表信息
     */
    private static SingleOutputStreamOperator<OrderWide> asyncJoinDimInfo(SingleOutputStreamOperator<OrderWide> orderWideDs) {
        // 异步关联维度信息1
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuInfo = asyncJoinDimInfo1(orderWideDs);
        // 异步关联维度信息2
        return asyncJoinDimInfo2(orderWideWithSpuInfo);
    }

    /**
     * 异步关联维度信息1
     *
     * @param orderWideDs 待进行维度关联的订单信息
     * @return 维度关联后的订单宽表信息
     */
    private static SingleOutputStreamOperator<OrderWide> asyncJoinDimInfo1(SingleOutputStreamOperator<OrderWide> orderWideDs) {

        // 1、关联用户信息
        SingleOutputStreamOperator<OrderWide> orderWideWithDimUserInfo = AsyncDataStream.unorderedWait(orderWideDs,
                new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                    @Override
                    public String getDimKey(OrderWide orderWide) {
                        return orderWide.getUser_id().toString();
                    }

                    @Override
                    public void setDimField(OrderWide orderWide, JSONObject dimJsonObj) {
                        // 设置性别
                        String gender = dimJsonObj.getString("GENDER");
                        orderWide.setUser_gender(gender);
                        // 设置年龄
                        String birthday = dimJsonObj.getString("BIRTHDAY");
                        int age = DateUtils.getAge(birthday, "yyyy-MM-dd");
                        orderWide.setUser_age(age);
                    }
                }, 60, TimeUnit.SECONDS);

        // 2、关联地区信息
        SingleOutputStreamOperator<OrderWide> orderWideWithDimProvinceInfo = AsyncDataStream.unorderedWait(orderWideWithDimUserInfo,
                new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getDimKey(OrderWide orderWide) {
                        return orderWide.getProvince_id().toString();
                    }

                    @Override
                    public void setDimField(OrderWide orderWide, JSONObject dimJsonObj) {
                        orderWide.setProvince_name(dimJsonObj.getString("NAME"));
                        orderWide.setProvince_iso_code(dimJsonObj.getString("ISO_CODE"));
                        orderWide.setProvince_area_code(dimJsonObj.getString("AREA_CODE"));
                        orderWide.setProvince_3166_2_code(dimJsonObj.getString("ISO_3166_2"));
                    }
                }, 60, TimeUnit.SECONDS);

        // 3、关联sku
        SingleOutputStreamOperator<OrderWide> orderWideWithSkuInfo = AsyncDataStream.unorderedWait(orderWideWithDimProvinceInfo,
                new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                    @Override
                    public String getDimKey(OrderWide orderWide) {
                        return orderWide.getSku_id().toString();
                    }

                    @Override
                    public void setDimField(OrderWide orderWide, JSONObject dimJsonObj) {
                        orderWide.setSku_name(dimJsonObj.getString("SKU_NAME"));
                        orderWide.setCategory3_id(dimJsonObj.getLong("CATEGORY3_ID"));
                        orderWide.setSpu_id(dimJsonObj.getLong("SPU_ID"));
                        orderWide.setTm_id(dimJsonObj.getLong("TM_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        // 4、关联spu
        return AsyncDataStream.unorderedWait(orderWideWithSkuInfo,
                new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public String getDimKey(OrderWide orderWide) {
                        return orderWide.getSpu_id().toString();
                    }

                    @Override
                    public void setDimField(OrderWide orderWide, JSONObject dimJsonObj) {
                        orderWide.setSpu_name(dimJsonObj.getString("SPU_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
    }

    /**
     * 异步关联维度信息2
     *
     * @param orderWideDs 待进行维度关联的订单信息
     * @return 维度关联后的订单宽表信息
     */
    private static SingleOutputStreamOperator<OrderWide> asyncJoinDimInfo2(SingleOutputStreamOperator<OrderWide> orderWideDs) {
        // 1、关联品类信息
        SingleOutputStreamOperator<OrderWide> orderWideWithCategoryInfo = AsyncDataStream.unorderedWait(orderWideDs,
                new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public String getDimKey(OrderWide orderWide) {
                        return orderWide.getCategory3_id().toString();
                    }

                    @Override
                    public void setDimField(OrderWide orderWide, JSONObject dimJsonObj) {
                        orderWide.setCategory3_name(dimJsonObj.getString("NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
        // 2、关联品牌信息
        return AsyncDataStream.unorderedWait(orderWideWithCategoryInfo,
                new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                    @Override
                    public String getDimKey(OrderWide orderWide) {
                        return orderWide.getTm_id().toString();
                    }

                    @Override
                    public void setDimField(OrderWide orderWide, JSONObject dimJsonObj) {
                        orderWide.setTm_name(dimJsonObj.getString("TM_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
    }

    /**
     * 将 String 转为 OrderInfo 类型，并设置创建时间(时间戳)
     */
    public static class MapOrderInfo extends RichMapFunction<String, OrderInfo> {

        private SimpleDateFormat sdf;

        @Override
        public void open(Configuration parameters) throws Exception {
            sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        }

        @Override
        public OrderInfo map(String value) throws Exception {
            OrderInfo orderInfo = JSON.parseObject(value, OrderInfo.class);
            long createTimestamp = sdf.parse(orderInfo.getCreate_time()).getTime();
            orderInfo.setCreate_ts(createTimestamp);
            return orderInfo;
        }
    }

    /**
     * 订单时间戳分配器
     */
    public static class OrderInfoTimestampAssigner implements SerializableTimestampAssigner<OrderInfo> {
        @Override
        public long extractTimestamp(OrderInfo orderInfo, long recordTimestamp) {
            return orderInfo.getCreate_ts();
        }
    }

    /**
     * 将 String 转为 OrderDetailInfo 类型，并设置创建时间(时间戳)
     */
    public static class MapOrderDetailInfo extends RichMapFunction<String, OrderDetail> {

        private SimpleDateFormat sdf;

        @Override
        public void open(Configuration parameters) throws Exception {
            sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        }

        @Override
        public OrderDetail map(String value) throws Exception {
            OrderDetail orderDetail = JSON.parseObject(value, OrderDetail.class);
            long createTimestamp = sdf.parse(orderDetail.getCreate_time()).getTime();
            orderDetail.setCreate_ts(createTimestamp);
            return orderDetail;
        }
    }

    /**
     * 订单详情时间戳分配器
     */
    private static class OrderDetailInfoTimestampAssigner implements SerializableTimestampAssigner<OrderDetail> {
        @Override
        public long extractTimestamp(OrderDetail orderDetail, long recordTimestamp) {
            return orderDetail.getCreate_ts();
        }
    }

    private static class OrderInfoProcessJoinFunction extends ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide> {

        @Override
        public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>.Context ctx, Collector<OrderWide> out) throws Exception {
            out.collect(new OrderWide(orderInfo, orderDetail));
        }
    }
}
